package bigdata36.damin;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class TableDataCURD2 {
	Table table = null;
	Connection connection = null;
	@Before
	public void init() throws Exception{
		Configuration config = HBaseConfiguration.create();
		connection = ConnectionFactory.createConnection(config);
		
		table= connection.getTable(TableName.valueOf("t_user"));
	}
	
	/**
	 * 单行查询   get
	 * @throws Exception 
	 */
	@Test
	public void getData() throws Exception{
		Get get = new Get(Bytes.toBytes("rk001"));
		//指定查哪些列，没有添加的列，查出来的数据都为空
		get.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
		Result result = table.get(get);
		byte[] value = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
		System.out.println(Bytes.toString(value));
		byte[] age = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("age"));
		System.out.println(Bytes.toString(age));
	}
	
	/**
	 * 多行查询
	 * @throws Exception 
	 * @throws Exception
	 */
	@Test
	public void testScanData() throws Exception{
		Scan scan = new Scan();
		//包含头不包含尾   //不设置的话，全表扫描
		scan.setStartRow(Bytes.toBytes("rk002"));
		scan.setStopRow(Bytes.toBytes("rk004"));
		
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.println(Bytes.toString(result.getRow()));
			byte[] value = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
			System.out.println(Bytes.toString(value));
			byte[] age = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("age"));
			System.out.println(Bytes.toString(age));
			System.out.println("-----------------------");
		}
	}
	
	/**
	 * 列值过滤器
	 * @throws Exception 
	 * @throws Exception
	 */
	@Test
	public void scanData() throws Exception{
		Scan scan = new Scan();
		//列值过滤器，   参数：1列簇名 2：列名   3：关系  4：值
		SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("base_info"), Bytes.toBytes("name"), CompareOp.EQUAL, Bytes.toBytes("zhangsan"));
		scan.setFilter(filter);
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.println(Bytes.toString(result.getRow()));
			byte[] value = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
			System.out.println(Bytes.toString(value));
			byte[] age = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("age"));
			System.out.println(Bytes.toString(age));
			System.out.println("-----------------------");
		}
	}
	/**
	 * 列名前缀过滤器
	 * @throws Exception 
	 */
	@Test
	public void testScanByColumn() throws Exception{
		Scan scan = new Scan();
		// 带有这个列名的都给过滤出来
		ColumnPrefixFilter filter = new ColumnPrefixFilter(Bytes.toBytes("name"));
		
		scan.setFilter(filter);
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.println(Bytes.toString(result.getRow()));
			byte[] value = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
			System.out.println(Bytes.toString(value));
			byte[] age = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("age"));
			System.out.println(Bytes.toString(age));
			System.out.println("-----------------------");
		}
	}
	
	/**
	 * rowkey 过滤器
	 * @throws IOException 
	 * @throws Exception
	 */
	@Test
	public void testScannerByRowkey() throws IOException{
		Scan scan = new Scan();
		//针对rowkey做过滤的
		RowFilter filter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("^rk"));
		
		scan.setFilter(filter);
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.println(Bytes.toString(result.getRow()));
			byte[] value = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
			System.out.println(Bytes.toString(value));
			byte[] age = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("age"));
			System.out.println(Bytes.toString(age));
			System.out.println("-----------------------");
		}
	}
	/**
	 * 多个过滤器联合使用
	 * @throws IOException 
	 */
	@Test
	public void testScanListFilter() throws IOException{
		Scan scan = new Scan();
		//过滤器集合     and   or
		FilterList list = new FilterList(Operator.MUST_PASS_ALL);
		
		RowFilter filter1 = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("^rk"));
		ColumnPrefixFilter filter2 = new ColumnPrefixFilter(Bytes.toBytes("name"));
		//把过滤器添加到集合中
		list.addFilter(filter1);
		list.addFilter(filter2);
		//把过滤器集合添加到scan
		scan.setFilter(list);
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.println(Bytes.toString(result.getRow()));
			byte[] value = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
			System.out.println(Bytes.toString(value));
			byte[] age = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("age"));
			System.out.println(Bytes.toString(age));
			System.out.println("-----------------------");
		}
	}
	
	
	@After
	public void close() throws Exception{
		table.close();
		connection.close();
	}
}
